package com.hyt.it.ogt.ks.config.canal;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.hyt.it.ogt.ks.util.RedisLock;

import lombok.extern.slf4j.Slf4j;

/****
 * canal客户端配置，用来监听canal服务器
 *
 * @author zhoumiao
 * @since 2022-11-09
 *
 */
@Slf4j
@Component
public class CanalMonitor implements CommandLineRunner {
    
    private static final String CANAL_MONITOR_THREAD_NAME = "canalMonitorThread";
    public static final String LOCK_KEY_HEAD = "canal-master-hander-key";
    public final static int LOCK_TIME = 60 * 5;

    private CanalConnector connector;
    /**
     * 是否关闭canal客户端
     */
    private volatile boolean isClosed = false;
    /**
     * canal客户端最大连接次数
     */
    private static final int max_connect_count = 5;
    /**
     * 当前已经连接的次数
     */
    private int currentConnectCount = 0;

    @Resource
    private CanalConfig canalConfig;
    @Resource
    private Dispatcher dispatcher;
    @Resource
    private RedisLock redisLock;

    
    @Override
    public void run(String... args) throws Exception {
        log.warn("======开始注册canal======");
        Thread thread = new Thread(() -> {
            initCanalConnection();
        }, CANAL_MONITOR_THREAD_NAME);
        thread.setDaemon(true);
        thread.start();
    }
    
    

    private void initCanalConnection() {
        String consumerId = UUID.randomUUID().toString();
        while (true) {
            try {
                // 获取执行权限
                if (!getCanalMasterHander(consumerId)) {
                    // 睡眠指定时间
                    TimeUnit.SECONDS.sleep(10);
                    // 这里不需要去释放连接，只是不去消费数据就可以了
                    // disconnect();
                } else {
                    // 给锁续命
                    if (continuationOfLifeMasterHander(consumerId)) {
                        if (null == connector) {
                            // 初始化连接
                            createConnection();
                            connect();
                        }
                        if (Thread.interrupted() || isClosed) {
                            log.warn("======canal客户端退出======");
                            break;
                        }
                        try {
                            // log.warn("# 考试消费数据");
                            // 获取指定数量的数据
                            Message message = connector.getWithoutAck(canalConfig.getBatchSize());
                            long batchId = message.getId();
                            int size = message.getEntries().size();
                            log.warn("# 考试消费数据,数据大小{}", size);
                            if (batchId > 0 && size != 0) {
                                handleDATAChange(message.getEntries());
                            }
                            // 提交确认
                            connector.ack(batchId);
                            // 睡眠指定时间
                            TimeUnit.SECONDS.sleep(canalConfig.getSleepSeconds());
                        } catch (InterruptedException e) {
                            log.error("canal线程中断报错：", e);
                            Thread.currentThread().interrupt();
                        } catch (Exception e) {
                            log.error("canal处理数据报错：", e);

                            // 连接次数是否已经大于最大连接次数 大于则退出
                            if (currentConnectCount >= max_connect_count) {
                                log.warn("======重连次数大于{} canal客户端退出======", currentConnectCount);
                                break;
                            }

                            // 重连处理
                            reconnect();
                        }
                    }
                }
            } catch (Exception e) {
                log.error("## canal连接报错", e);
            }
        }
    }

    private void createConnection() {
        //在canal部署的conf/canal.properties ip和端口信息
        connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalConfig.getCanalip(), canalConfig.getCanalport()), canalConfig.getCanaldestination(), "", "");
    }



    private boolean getCanalMasterHander(String consumerId) {
        // canal 消费者只能一个进行消费，不然数据会乱 这里提供一个开关
        Boolean handerResult = redisLock.tryLock(LOCK_KEY_HEAD, consumerId, LOCK_TIME);
        log.warn("# canal获取主节点执行权限:{}", handerResult);
        return handerResult;
    }
    
    private boolean continuationOfLifeMasterHander(String consumerId) {
        // log.warn("# 开始给锁{}续命", LOCK_KEY_HEAD);
        Boolean continuationOfLifeLock = redisLock.continuationOfLifeLock(LOCK_KEY_HEAD, consumerId, LOCK_TIME);
        log.warn("# 续命结果{}", continuationOfLifeLock);
        return continuationOfLifeLock;
    }

    private void reconnect() {
        log.warn("======canal重连======");
        disconnect();
        connect();
    }

    private void connect() {
        // 打开连接
        connector.connect();
        // 订阅数据库的数据表，并对数据表进行过滤监听
        connector.subscribe(canalConfig.getDatabase());
        currentConnectCount++;
        isClosed = false;
        log.info("======canal启动完成======");
    }

    private void handleDATAChange(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            // 只解析mysql事务的操作，其他的不解析
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            // RowChange对象，包含了一行数据变化的所有特征
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            // 获取当前操作所属的数据库
            String dbName = entry.getHeader().getSchemaName();
            // 获取当前操作所属的表
            String tableName = entry.getHeader().getTableName();

            // 事务提交时间
            long timestamp = entry.getHeader().getExecuteTime();


            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                try {
                    dispatcher.dispatch(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), dbName, tableName, eventType, timestamp);
                } catch (Exception e) {
                    log.error("处理具体数据报错：参数【{} {} {} {} {} {}】 错误【{}】", rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), dbName, tableName, eventType, timestamp, e);
                }
            }
        }
    }

    public void close() {
        isClosed = true;
    }

    private void disconnect() {
        if (Objects.nonNull(connector)) {
            try {
                log.info("======关闭canal======");
                // 回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
                connector.rollback();
                connector.unsubscribe();
                connector.disconnect();
            } catch (Exception e) {
                log.error("关闭canal报错：", e);
            }
        }
    }

}

